package com.canal.config;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.common.utils.AddressUtils;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Bean;
import org.springframework.scheduling.annotation.Async;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Description
 * @Author 顾志杰
 * @Date 2022/8/7 16:39
 */
@Component
@Slf4j
public class CanalConfig {



    @PostConstruct
    public void init() {
        new Thread(() -> {
          CanalConnector connector = CanalConnectors.newSingleConnector(
                  new InetSocketAddress(AddressUtils.getHostIp(), 11111), "example", "", "");
          try {
            connector.connect();
            connector.subscribe(".*\\..*");
            connector.rollback();
            while (true) {
              //一回取多少数据
              Message message = connector.getWithoutAck(100);
              long batchId = message.getId();
              if (batchId == -1 || message.getEntries().size() == 0) {
                log.info("没有数据更新!!!!!!!!!");
                Thread.sleep(1000);
              } else {
                CanalConfig.this.printEntry(message.getEntries());
              }
              connector.ack(batchId);
            }
          } catch (InterruptedException e) {
            throw new RuntimeException(e);
          } finally {
            connector.disconnect();
          }
        },"canal").start();

    }

    private void printEntry(List<CanalEntry.Entry> entries) {
        for (CanalEntry.Entry entry : entries) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }
            CanalEntry.RowChange rowChange = null;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("data:" + entry.toString(), e);
            }
            CanalEntry.EventType eventType = rowChange.getEventType();
            log.info(String.format("=======================&gt; binlog[%s:%s] , name[%s:%s], eventType : %s",
                    entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                    entry.getHeader().getSchemaName(), entry.getHeader().getTableName(),
                    eventType));

            for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
                log.info("=========================================");
                log.info("event type " + eventType.name());
                log.info("****************************** before");
                printColumn(rowData.getBeforeColumnsList());
                log.info("****************************** after");
                printColumn(rowData.getAfterColumnsList());
            }
        }
    }

    private void printColumn(List<CanalEntry.Column> columns) {
        if (CollectionUtils.isEmpty(columns)) {
            return;
        }
        for (CanalEntry.Column column : columns) {
            log.info(column.getName() + ":" + column.getValue() + "   update=" + column.getUpdated());
        }
    }

}
